package com.sg.java.apps;

import com.google.common.collect.Lists;
import com.google.gson.JsonObject;
import com.sg.java.HBaseNamespace;
import com.sg.java.MsgFromKafkaToHBaseTransformer;
import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import com.sg.java.util.HBaseUtils;
import com.sg.java.util.JSONUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

import static org.apache.hadoop.hbase.util.Bytes.toBytes;

public class KafkaConsumerClient {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerClient.class);

    public static final Supplier<MsgFromKafkaToHBaseTransformer> cms_volt_curve_transformer_generator = () -> {
        String namespace = HBaseNamespace.CQ_HBASE_NAMESPACE;
        String topic     = "cms_volt_curve";
        String rowKey    = "row";
        return MsgFromKafkaToHBaseTransformer.builder()
                                             .namespace(namespace)
                                             .topic(topic)
                                             .tableDescriptor(tableName -> TableDescriptorBuilder
                                                     .newBuilder(tableName)
                                                     .setColumnFamilies(Lists.newArrayList(
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("info"))
                                                                                          .setMaxVersions(3)
                                                                                          .build(),
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("U"))
                                                                                          .setMaxVersions(3)
                                                                                          .build()))
                                                     .build())
                                             .etl((jo) -> {
                                                 if (!JSONUtils.jsonElementIsNotBlank(jo.get("U"))) {
                                                     jo.addProperty("U", "");
                                                 }
                                                 return JSONUtils.jsonElementIsNotBlank(jo.get("METER_ID")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("DATA_DATE")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("PHASE_FLAG")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("COL_TIME")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("ORG_NO"));

                                             })
                                             .getDataDate(jo -> jo.get("DATA_DATE").getAsString())
                                             .rowGenerator((jo) -> Pair.of(rowKey, jo.get("METER_ID")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("DATA_DATE")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("PHASE_FLAG").getAsString()))
                                             .handleAndMapToPut(jo -> {
                                                 String rowValue = jo.get(rowKey).getAsString();
                                                 Put    put      = new Put(toBytes(rowValue));
                                                 String column_COL_TIME_U = jo.get("COL_TIME")
                                                                              .getAsString()
                                                                              .substring(8);
                                                 put.addColumn(toBytes("info"), toBytes("METER_ID"), toBytes(jo.get("METER_ID")
                                                                                                               .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("DATA_DATE"), toBytes(jo.get("DATA_DATE")
                                                                                                                .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("PHASE_FLAG"), toBytes(jo.get("PHASE_FLAG")
                                                                                                                 .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("ORG_NO"), toBytes(jo.get("ORG_NO")
                                                                                                             .getAsString()));
                                                 put.addColumn(toBytes("U"), toBytes(column_COL_TIME_U), toBytes(jo.get("U")
                                                                                                                   .getAsString()));
                                                 return put;
                                             })
                                             .build();
    };

    public static final Supplier<MsgFromKafkaToHBaseTransformer> cms_cur_curve_transformer_generator = () -> {
        String namespace = HBaseNamespace.CQ_HBASE_NAMESPACE;
        String topic     = "cms_cur_curve";
        String rowKey    = "row";
        return MsgFromKafkaToHBaseTransformer.builder()
                                             .namespace(namespace)
                                             .topic(topic)
                                             .tableDescriptor(tableName -> TableDescriptorBuilder
                                                     .newBuilder(tableName)
                                                     .setColumnFamilies(Lists.newArrayList(
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("info"))
                                                                                          .setMaxVersions(3)
                                                                                          .build(),
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("I"))
                                                                                          .setMaxVersions(3)
                                                                                          .build()))
                                                     .build())
                                             .etl((jo) -> {
                                                 if (!JSONUtils.jsonElementIsNotBlank(jo.get("I"))) {
                                                     jo.addProperty("I", "");
                                                 }
                                                 return JSONUtils.jsonElementIsNotBlank(jo.get("METER_ID")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("DATA_DATE")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("PHASE_FLAG")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("COL_TIME")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("ORG_NO"));
                                             })
                                             .getDataDate(jo -> jo.get("DATA_DATE").getAsString())
                                             .rowGenerator((jo) -> Pair.of(rowKey, jo.get("METER_ID")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("DATA_DATE")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("PHASE_FLAG").getAsString()))
                                             .handleAndMapToPut(jo -> {
                                                 String rowValue = jo.get(rowKey).getAsString();
                                                 Put    put      = new Put(toBytes(rowValue));
                                                 String column_COL_TIME_U = jo.get("COL_TIME")
                                                                              .getAsString()
                                                                              .substring(8);
                                                 put.addColumn(toBytes("info"), toBytes("METER_ID"), toBytes(jo.get("METER_ID")
                                                                                                               .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("DATA_DATE"), toBytes(jo.get("DATA_DATE")
                                                                                                                .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("PHASE_FLAG"), toBytes(jo.get("PHASE_FLAG")
                                                                                                                 .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("ORG_NO"), toBytes(jo.get("ORG_NO")
                                                                                                             .getAsString()));
                                                 put.addColumn(toBytes("I"), toBytes(column_COL_TIME_U), toBytes(jo.get("I")
                                                                                                                   .getAsString()));
                                                 return put;
                                             })
                                             .build();
    };

    public static final Supplier<MsgFromKafkaToHBaseTransformer> cms_power_curve_transformer_generator = () -> {
        String namespace = HBaseNamespace.CQ_HBASE_NAMESPACE;
        String topic     = "cms_power_curve";
        String rowKey    = "row";
        return MsgFromKafkaToHBaseTransformer.builder()
                                             .namespace(namespace)
                                             .topic(topic)
                                             .tableDescriptor(tableName -> TableDescriptorBuilder
                                                     .newBuilder(tableName)
                                                     .setColumnFamilies(Lists.newArrayList(
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("info"))
                                                                                          .setMaxVersions(3)
                                                                                          .build(),
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("P"))
                                                                                          .setMaxVersions(3)
                                                                                          .build()))
                                                     .build())
                                             .etl((jo) -> {
                                                 if (!JSONUtils.jsonElementIsNotBlank(jo.get("P"))) {
                                                     jo.addProperty("P", "");
                                                 }
                                                 return JSONUtils.jsonElementIsNotBlank(jo.get("METER_ID")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("DATA_DATE")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("DATA_TYPE")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("COL_TIME")) &&
                                                        JSONUtils.jsonElementIsNotBlank(jo.get("ORG_NO"));
                                             })
                                             .getDataDate(jo -> jo.get("DATA_DATE").getAsString())
                                             .rowGenerator((jo) -> Pair.of(rowKey, jo.get("METER_ID")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("DATA_DATE")
                                                                                     .getAsString() + "-" +
                                                                                   jo.get("DATA_TYPE").getAsString()))
                                             .handleAndMapToPut(jo -> {
                                                 String rowValue = jo.get(rowKey).getAsString();
                                                 Put    put      = new Put(toBytes(rowValue));
                                                 String column_COL_TIME_U = jo.get("COL_TIME")
                                                                              .getAsString()
                                                                              .substring(8);
                                                 put.addColumn(toBytes("info"), toBytes("METER_ID"), toBytes(jo.get("METER_ID")
                                                                                                               .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("DATA_DATE"), toBytes(jo.get("DATA_DATE")
                                                                                                                .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("DATA_TYPE"), toBytes(jo.get("DATA_TYPE")
                                                                                                                .getAsString()));
                                                 put.addColumn(toBytes("info"), toBytes("ORG_NO"), toBytes(jo.get("ORG_NO")
                                                                                                             .getAsString()));
                                                 put.addColumn(toBytes("P"), toBytes(column_COL_TIME_U), toBytes(jo.get("P")
                                                                                                                   .getAsString()));
                                                 return put;
                                             })
                                             .build();
    };


    //1.2.3. 低压设备遥测断面信息
    public static final Supplier<MsgFromKafkaToHBaseTransformer> dms_decloud_mlv_measuration_transformer_generator = () -> {
        String namespace = HBaseNamespace.CQ_HBASE_NAMESPACE;
        String topic     = "dms_decloud_mlv_measuration";
        String rowKey    = "row";
        return MsgFromKafkaToHBaseTransformer.builder()
                                             .namespace(namespace)
                                             .topic(topic)
                                             .tableDescriptor(tableName -> TableDescriptorBuilder
                                                     .newBuilder(tableName)
                                                     .setColumnFamilies(Lists.newArrayList(
                                                             ColumnFamilyDescriptorBuilder.newBuilder(toBytes("info"))
                                                                                          .setMaxVersions(3)
                                                                                          .build()))
                                                     .build())
                                             .etl(jo -> {
                                                 if (jo == null) {
                                                     return false;
                                                 }
                                                 final JsonObject measurentValueList = jo.getAsJsonObject("MeasurentValueList");
                                                 return measurentValueList != null;
                                             })
                                             .getDataDate(jo -> jo.get("timeStamp")
                                                                  .getAsString()
                                                                  .substring(0, 10)
                                                                  .replaceAll("-", ""))
                                             .rowGenerator(jo ->
                                                           {
                                                               JsonObject measurentValueList = jo.getAsJsonObject("MeasurentValueList");
                                                               String mRId = measurentValueList.get("mRId")
                                                                                               .getAsString();
                                                               String time = measurentValueList.get("time")
                                                                                               .getAsString();
                                                               return Pair.of(rowKey, mRId + "_" + time);
                                                           }
                                                          )
                                             .handleAndMapToPut(jo -> {
                                                 String     rowValue           = jo.get(rowKey).getAsString();
                                                 Put        put                = new Put(toBytes(rowValue));
                                                 JsonObject measurentValueList = jo.getAsJsonObject("MeasurentValueList");
                                                 String time = measurentValueList.get("time")
                                                                                 .getAsString();
                                                 String equipType = measurentValueList.get("equipType")
                                                                                      .getAsString();
                                                 String areaType = measurentValueList.get("areaType")
                                                                                     .getAsString();
                                                 String mRId = measurentValueList.get("mRId")
                                                                                 .getAsString();
                                                 String analog = measurentValueList.getAsJsonArray("Analog").toString();

                                                 put.addColumn(toBytes("info"), toBytes("time"), toBytes(time));
                                                 put.addColumn(toBytes("info"), toBytes("equipType"), toBytes(equipType));
                                                 put.addColumn(toBytes("info"), toBytes("areaType"), toBytes(areaType));
                                                 put.addColumn(toBytes("info"), toBytes("mRId"), toBytes(mRId));
                                                 put.addColumn(toBytes("info"), toBytes("Analog"), toBytes(analog));
                                                 return put;
                                             })
                                             .build();
    };

    public static final List<MsgFromKafkaToHBaseTransformer> TRANSFORMERS = Lists.newArrayList(
            dms_decloud_mlv_measuration_transformer_generator.get(),
            cms_volt_curve_transformer_generator.get(),
            cms_cur_curve_transformer_generator.get(),
            cms_power_curve_transformer_generator.get()

                                                                                              );

    private static final int eachTransformerConsumersNumber = 1;

    private static final int POLL_TIMEOUT_S = 30;

    public static String kafkaMsgPrint(ConsumerRecord<String, String> cr) {
        return "topic:" + cr.topic() + "\t" +
               "partition:" + cr.partition() + "\t" +
               "offset:" + cr.offset() + "\t" +
               "value:" + cr.value();
    }

    public static void main(String[] args) throws Exception {
        log.info("订阅主题：{}", TRANSFORMERS);
        final int             consumerThreadPoolSize = TRANSFORMERS.size() * eachTransformerConsumersNumber;
        final ExecutorService executors              = Executors.newFixedThreadPool(consumerThreadPoolSize);
        log.info("创建消费者线程池，size={}", consumerThreadPoolSize);
        log.info("读取kafka消费者配置文件");
        final Properties prop = PropertiesUtil.createPropertiesFromResource(ResourcePath.kafka_consumer_properties);
        log.info("kafka消费者配置文件内容：" + prop);

        log.info("kafka服务端地址bootstrap.servers：{}", prop.getProperty("bootstrap.servers"));
        log.info("消费者组group.id：{}", prop.getProperty("group.id"));
        log.info("单次消息拉取最大等待时间poll.timeout：{}s", POLL_TIMEOUT_S);
        final int maxPollRecords = Integer.parseInt(prop.getProperty("max.poll.records", "50"));
        log.info("单次消息拉取数据条数max.poll.records:{}", maxPollRecords);

        log.info("topics在hbase中若不存在表则进行建表，若存在表则跳过");
        //hbase connection 是线程安全的，非常重量级，全局维护一个就可以，并且连接建立昂贵，同时建立过多连接会导致hbase server拒绝
        final Connection globalConn = HBaseUtils.getHBaseConn(PropertiesUtil.createPropertiesFromResource(ResourcePath.hbase_properties));
        TRANSFORMERS.parallelStream().forEach(transformer -> {
            for (int i = 0; i < eachTransformerConsumersNumber; i++) {
                String consumerName = "consumer" + i + "_" + transformer.getTopic() + "_" + Thread.currentThread()
                                                                                                  .getId();
                executors.execute(() -> {
                    try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
                         final Admin admin = globalConn.getAdmin()) {
                        log.info(consumerName + "订阅topic：{}", transformer.getTopic());
                        consumer.subscribe(Collections.singleton(transformer.getTopic()));
                        long startS       = System.currentTimeMillis();
                        long tempS        = System.currentTimeMillis();
                        long totalPoll    = 0;
                        long totalValid   = 0;
                        long totalInvalid = 0;
                        long lastValid    = 0;
                        long lastInvalid  = 0;
//                        List<JsonObject> tempInvalid = new ArrayList<>();
                        Map<String, Pair<Table, List<Put>>> putsMap = new HashMap<>();
                        while (true) {
                            ConsumerRecords<String, String> records      = consumer.poll(Duration.ofSeconds(POLL_TIMEOUT_S));
                            ConsumerRecord<String, String>  newestRecord = null;
                            if (!records.isEmpty()) {
                                for (ConsumerRecord<String, String> record : records) {
                                    newestRecord = record;
                                    JsonObject jo = JSONUtils.toJsonObject(record.value());
                                    if (jo != null) {
                                        totalPoll++;
                                        if (transformer.getEtl().test(jo)) {
                                            totalValid++;
                                            lastValid++;
                                            Pair<String, String> kv = transformer.getRowGenerator().apply(jo);
                                            jo.addProperty(kv.getKey(), kv.getValue());
                                            //要入的hbase表名
                                            String inputTableName = transformer.getNamespace() + ":" + transformer.getTopic() + "_" + transformer.getGetDataDate()
                                                                                                                                                 .apply(jo);
                                            Pair<Table, List<Put>> pair = putsMap.get(inputTableName);
                                            if (pair == null) {
                                                TableName tn = TableName.valueOf(inputTableName);
                                                if (!admin.isTableAvailable(tn)) {
                                                    //此处能用字符串加锁，但必须用常量池的字符串，因为该字符串变量存在引用拼接，会在堆中new，如果2个内容相同的字符串都是来自堆中的不同对象，则锁异常，所以用intern先去常量池取，有就返回，没有则存入字符串常量池并返回
                                                    synchronized (inputTableName.intern()) {
                                                        if (!admin.isTableAvailable(tn)) {
                                                            admin.createTable(transformer.getTableDescriptor()
                                                                                         .apply(tn));
                                                        }
                                                    }
                                                }
                                                Table     table = globalConn.getTable(tn);
                                                List<Put> puts  = new ArrayList<>(maxPollRecords);
                                                pair = Pair.of(table, puts);
                                                putsMap.put(inputTableName, pair);
                                            }
                                            pair.getRight().add(transformer.getHandleAndMapToPut().apply(jo));
                                        }
                                        else {
                                            totalInvalid++;
                                            lastInvalid++;
//                                            tempInvalid.add(jo);
                                        }
                                    }
                                }
                                putsMap.forEach((k, v) -> {
                                    try {
                                        if (!v.getRight().isEmpty()) {
                                            //hbase不同表数据分别入库
                                            v.getLeft().put(v.getRight());
                                            //清空put列表
                                            v.getRight().clear();
                                        }
                                    }
                                    catch (IOException e) {
                                        throw new RuntimeException(e);
                                    }
                                });
                                //入库成功手动同步提交偏移量
                                consumer.commitSync();
                            }
                            //运行超过5分钟，输出下日志
                            if (System.currentTimeMillis() - tempS >= 5 * 60 * 1000) {
                                if (newestRecord != null) {
                                    log.info(consumerName + "\t最新的数据记录：{}", newestRecord);
                                }
                                log.info(consumerName + "\t开始时间戳：{}\t已运行总时间：{}ms\t已拉取总数：{}\t有效总数：{}\t无效总数：{}\t最近半小时有效和无效数据：{}/{}",
                                         startS,
                                         System.currentTimeMillis() - startS,
                                         totalPoll,
                                         totalValid,
                                         totalInvalid,
                                         lastValid,
                                         lastInvalid
                                        );
//                                tempInvalid.clear();
                                lastValid   = 0;
                                lastInvalid = 0;
                                tempS       = System.currentTimeMillis();
//                                if (!tempInvalid.isEmpty()) {
//                                    //日志刷太多了，先不输出
//                                    StringJoiner sj = new StringJoiner("\n");
//                                    tempInvalid.forEach(jo -> sj.add(jo.toString()));
//                                    log.info(consumerName + "异常数据size：{}", tempInvalid.size());
//                                    log.info(sj.toString());
//                                    tempInvalid.clear();
//                                } else {
//                                    log.info(consumerName + "暂无异常数据");
//                                }
                            }
                        }
                    }
                    catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
                log.info(consumerName + "运行中...");
            }
        });
        log.info("所有消费者运行中...");
    }


}
